src/parseable/streams.rs (1)

345-353: ⚠️ Potential issue

File::created() will panic on common Linux filesystems

std::fs::Metadata::created() is not implemented on most Unix filesystems and returns an error.
The expect("Creation time should be accessible") will therefore bring the whole conversion task – and potentially the server – down on virtually every Linux deployment.

Replace with a fall-back that relies on the consistently available modified() timestamp (or skip the filtering when created() is unsupported).

- let creation = path.metadata()?.created().expect("Creation time should be accessible");
+ let creation = path
+     .metadata()?
+     .created()
+     .or_else(|_| path.metadata()?.modified())?; // fallback

src/handlers/http/modal/server.rs (1)

126-131: Startup sync duplication

Same concern as in query_server.rs: this detached sync_start() may overlap with the periodic sync::handler spawned just below.
Verify that the two paths cannot run concurrently or gate them behind a single-execution guard.

src/handlers/http/modal/query_server.rs (1)

131-136: Possible duplicate & unmanaged sync task

sync_start() is fired here while a separate periodic sync thread (sync::handler) is launched a few lines later.
Without a guard this can lead to two independent object-store uploads running simultaneously, wasting IO and increasing race-surface.
Consider:

1. Running the startup sync inside the periodic worker before its main loop begins, or
2. Protecting concurrent runs with a mutex/atomic “sync-in-progress” flag, and/or
3. At least keep the JoinHandle so it can be awaited during shutdown.

src/parseable/staging/reader.rs (1)

88-103: Great improvement, but one remaining unwrap still panics

The new match File::open keeps the reader resilient – nice.
However the earlier file.metadata().unwrap() (line 51) will still crash the entire process for an I/O error (e.g. permission issue).
Replace with a fallible path similar to this block and skip the file on failure.

src/handlers/http/health_check.rs (1)

63-65: Hard-coded 5 s sleep may stall graceful shutdown

A fixed sleep(5s) delays shutdown even when no in-flight requests exist and also blocks all new requests (flag is already set).
Consider making this configurable or replacing with a loop that waits until active-request-counter drops to zero (with max timeout).

src/parseable/streams.rs (2)

70-77: arrow_path_to_parquet argument order is confusing

staging_path is expected to be a directory, yet tests pass a file path.
Consider tightening the contract:

-fn arrow_path_to_parquet(staging_path: &Path, path: &Path, random_string: &str)
+fn arrow_path_to_parquet(staging_dir: &Path, arrow_file: &Path, random_string: &str)

and add an early debug_assert!(staging_dir.is_dir()) to prevent accidental misuse.
This will avoid silently generating paths like /some/file.arrows/<name>.parquet.

---

588-604: Metric update panics on I/O errors

file.metadata().unwrap() will panic on any transient I/O failure.
Prefer if let Ok(meta) = file.metadata() to keep the sync path resilient under load.

src/sync.rs (1)

155-161: JoinSet drain duplicates log code – factor into helper

The same success / failure logging block appears three times in this file.
Extracting a small helper (log_join_result(res, "object store sync")) will remove repetition and keep future fixes in one spot.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

src/sync.rs (4)

131-144: Potential starvation & duplicated code around JoinSet loop

Every tick builds a fresh JoinSet, fires sync_all_streams, then blocks until all joins resolve.
If a large backlog exists, the loop may overrun the next tick, defeating periodic behaviour.

Consider extracting the common “spawn, join, log” logic used here and in local_sync into a reusable helper that:

1. Re-uses a single JoinSet (or a bounded one) across ticks, or
2. Offloads joins into a detached task so the driving loop never stalls.

This will keep upload cadence predictable and DRY-up the nearly identical code in the two sync paths.

---

195-207: Duplicate flush/convert scaffold – factor out to avoid divergence

The block mirrors lines 131-144 almost verbatim (different callback only). If a bug fix is needed later you’ll have to patch two places.

Extract into something like:

async fn run_and_log<F>(name: &str, threshold: Duration, f: F)
where
    F: FnOnce(&mut JoinSet<()>) + Send + 'static,
{ … }

and call it from both locations.

---

238-264: sync_start runs heavy IO on the main thread during startup

sync_start() performs full flush + upload synchronously before returning. On large datasets this can delay server readiness and make liveness probes fail.

Introduce either:

• a “started but warming up” state exposed via health-check, or
• run this in the background and let the server accept traffic immediately (with proper back-pressure).

Also consider making the 15 s threshold configurable – real-world clusters may legitimately take longer.

---

269-278: Minor: inconsistent log levels

log_join_result() logs success with info!, failures with warn!, and join errors with error!.
You may want to promote task failures (Ok(Err(_))) to error! as well to ease alerting.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

src/sync.rs (1)

131-145: Consider graceful early-exit for still-running JoinSet tasks

If inbox_rx receives a shutdown signal while sync_all_streams is in progress, any in-flight uploads inside joinset keep running until completion because the JoinSet lives only inside this closure.
In practice that is fine, but you lose the option to abort quickly during shutdown or when the monitor times out.

A lightweight pattern is to call joinset.shutdown().await; before the while let Some(..) loop, which causes unfinished tasks to be aborted and immediately joined:

-                                let mut joinset = JoinSet::new();
+                                let mut joinset = JoinSet::new();

                                 sync_all_streams(&mut joinset);

+                                // Abort remaining tasks if the outer future is cancelled
+                                joinset.shutdown().await;
                                 while let Some(res) = joinset.join_next().await {
                                     log_join_result(res, "object store sync");
                                 }

This avoids dangling uploads during shutdown and keeps task lifetimes explicit.

src/parseable/streams.rs (1)

380-387: Cross-device rename can fail – fall back to copy/remove

fs::rename fails with EXDEV if the staging disk and the in-process directory reside on
different mounts (common in containerised setups with bind-mounts).
Handle that case explicitly:

-                    if let Err(e) = fs::rename(&arrow_file_path, &new_path) {
+                    if let Err(e) = fs::rename(&arrow_file_path, &new_path) {
                         error!(
                             "Failed to rename arrow file to inprocess directory: {} -> {}: {e}",
                             arrow_file_path.display(),
                             new_path.display()
                         );
+                        if e.kind() == std::io::ErrorKind::CrossDeviceLink {
+                            if let Err(copy_err) = fs::copy(&arrow_file_path, &new_path) {
+                                error!("Copy fallback also failed: {copy_err}");
+                            } else {
+                                let _ = remove_file(&arrow_file_path);
+                        }
                     }

This prevents the entire batch from stalling on heterogeneous storage layouts.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/sync.rs:78-83
Timestamp: 2025-06-16T04:56:21.596Z
Learning: In Rust async code, `#[tokio::main]` is appropriate when functions are called from separate OS threads (like `thread::spawn()`), as it creates a new Tokio runtime for that thread. The "Cannot start a runtime from within a runtime" error only occurs when `#[tokio::main]` functions are called from within an existing Tokio runtime context.

src/parseable/streams.rs (2)

69-84: arrow_path_to_parquet trusts stream_staging_path is a directory

parquet_path.push(filename_with_random_number) silently produces paths like
/tmp/myfile.arrows/<generated>.parquet when stream_staging_path is accidentally a
file (the current unit-tests pass such a path).
At runtime this is unlikely, but if it happens the rename that follows will fail with
Not a directory.

-    let mut parquet_path = stream_staging_path.to_owned();
-    parquet_path.push(filename_with_random_number);
+    // Guard against a file being passed instead of the staging directory.
+    let mut parquet_path = if stream_staging_path.is_dir() {
+        stream_staging_path.to_path_buf()
+    } else {
+        stream_staging_path
+            .parent()
+            .unwrap_or_else(|| Path::new("."))
+            .to_path_buf()
+    };
+    parquet_path.push(filename_with_random_number);

This keeps the function robust and lets the unit-tests keep their current call-site.

---

585-593: Metrics reset only clears “arrows”/“parquet” buckets

The helper zeroes STAGING_FILES and both STORAGE_SIZE gauges, but only for the
labels "arrows" and "parquet". If a future extension introduces another label
(e.g. "orc"), the metrics will become stale. Consider looping through
metrics::STORAGE_SIZE.collect() and clearing every entry for the current stream.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/parseable/streams.rs:351-355
Timestamp: 2025-06-16T05:20:18.557Z
Learning: In the Parseable codebase, arrow files are expected to always have valid creation or modified timestamps as a basic system assumption. The conversion flow uses expect() on file metadata operations to enforce this invariant with fail-fast behavior rather than graceful error handling, as violations represent fundamental system issues that should cause immediate failures.

src/parseable/streams.rs (1)

687-706: Opening the .part file with append(true) is still fragile
(see previous discussion – unchanged)

If a stale .part file survives a crash, the next run will silently append new
bytes, producing an invalid parquet footer. Using truncate(true) avoids this
edge case at zero cost.

-    let mut part_file = OpenOptions::new()
-        .create(true)
-        .append(true)
+    let mut part_file = OpenOptions::new()
+        .write(true)
+        .create(true)
+        .truncate(true)

src/sync.rs (1)

119-145: Avoid the extra tokio::spawn indirection – run the interval loop directly

Spawning a second task inside the already-spawned handle adds an unnecessary layer of scheduling, makes panic/ cancellation semantics harder to reason about, and forces the surrounding catch_unwind dance.

You can drop the inner tokio::spawn and execute the interval loop in the outer task itself:

-    let result = tokio::spawn(async move {
-        let mut sync_interval = interval_at(next_minute(), STORAGE_UPLOAD_INTERVAL);
-        ...
-    });
-
-    match AssertUnwindSafe(result).catch_unwind().await {
+    let result = AssertUnwindSafe(async move {
+        let mut sync_interval = interval_at(next_minute(), STORAGE_UPLOAD_INTERVAL);
+        ...
+    })
+    .catch_unwind()
+    .await;
+
+    match result {
         Ok(_) => { /* handled below */ }

This keeps the same panic protection while removing one level of indirection.

src/parseable/streams.rs (3)

72-86: arrow_path_to_parquet first parameter is a directory, but tests pass a file path

stream_staging_path is expected to be the stream’s staging directory (self.data_path).
Passing the full arrow file path (as the unit-tests currently do) yields a nonsensical output like:

/tmp/arrow_file_path/converted_name.parquet

Consider renaming the parameter to staging_dir and fixing the tests to call:

arrow_path_to_parquet(file_path.parent().unwrap(), &file_path, random)

to reflect real-world usage.

---

210-223: arrow_files() does an O(n) metadata stat per file every tick

Calling metadata() for each entry on every sync cycle can become costly when tens-of-thousands of files accumulate.

If ordering by mtime is not strictly required, drop the sort & metadata hit.
If ordering is needed, cache the (PathBuf, SystemTime) vector once per run of convert_disk_files_to_parquet instead of on every directory scan.

---

263-288: Single minute directory may mix older arrow files – revisit atomicity guarantee

arrow_files_grouped_exclude_time moves all eligible files—regardless of their
original minute—into a directory named processing_<current-minute-1>.
Files that are >1 minute old therefore get grouped together, breaking the
“one-minute-per-conversion-task” invariant described in the design docs.

Either create the directory using each file’s own minute or keep per-minute
sub-directories when doing the move.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/sync.rs:78-83
Timestamp: 2025-06-16T04:56:21.596Z
Learning: In Rust async code, `#[tokio::main]` is appropriate when functions are called from separate OS threads (like `thread::spawn()`), as it creates a new Tokio runtime for that thread. The "Cannot start a runtime from within a runtime" error only occurs when `#[tokio::main]` functions are called from within an existing Tokio runtime context.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/parseable/streams.rs:351-355
Timestamp: 2025-06-16T05:20:18.557Z
Learning: In the Parseable codebase, arrow files are expected to always have valid creation or modified timestamps as a basic system assumption. The conversion flow uses expect() on file metadata operations to enforce this invariant with fail-fast behavior rather than graceful error handling, as violations represent fundamental system issues that should cause immediate failures.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1346
File: src/parseable/streams.rs:680-686
Timestamp: 2025-06-16T05:30:13.360Z
Learning: The parquet conversion system uses a minute-based atomic design where each conversion task operates on different sets of arrow files moved to timestamped "inprocess_" directories, ensuring no conflicts between tasks and no retry mechanisms that could cause file corruption.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro